package com.ferry.id.snowflake;

import com.alibaba.fastjson.JSON;
import com.ferry.id.common.PropertyFactory;
import com.google.common.collect.Maps;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.io.FileUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * Created by daiyong
 */
public class ZkHolder {
	private static final Logger LOGGER = LoggerFactory.getLogger(ZkHolder.class);

	private String zkAddressNode = null;
	private String listenAddress = null;
	private int workerID;
	private static final String PREFIX_ZK_PATH = "/snowflake/" + PropertyFactory.getProperties().getProperty("ferry-id.name");
	private static final String PROP_PATH = System.getProperty("java.io.tmpdir") + File.separator + PropertyFactory.getProperties().getProperty("ferry-id.name") + "/%s/workerID.properties";
	private static final String PATH_FOREVER = PREFIX_ZK_PATH + "/forever";//保存所有数据持久的节点
	private String ip;
	private String port;
	private String connectionString;
	private long lastUpdateTime;

	public ZkHolder(String ip, String port, String connectionString) {
		this.ip = ip;
		this.port = port;
		this.listenAddress = ip + ":" + port;
		this.connectionString = connectionString;
	}

	public boolean init() {
		try {
			//链接zk
			CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().connectString(connectionString)
					.retryPolicy(new RetryUntilElapsed(1000, 4))
					.connectionTimeoutMs(10000)
					.sessionTimeoutMs(6000)
					.build();
			curatorFramework.start();

			Stat stat = curatorFramework.checkExists().forPath(PATH_FOREVER);

			//第一次启动
			if (null == stat) {
				zkAddressNode = curatorFramework
						.create()
						.creatingParentsIfNeeded()
						.withMode(CreateMode.PERSISTENT_SEQUENTIAL)
						.forPath(PATH_FOREVER + "/" + listenAddress + "-", buildData().getBytes());
				updateLocalWorkerID(workerID);
				scheduledUploadData(curatorFramework, zkAddressNode);
			}
			//不是第一次启动
			else {
				Map<String, Integer> nodeMap = Maps.newHashMap();//ip:port->00001
				Map<String, String> realNode = Maps.newHashMap();//ip:port->(ipport-000001)
				//存在根节点,先检查是否有属于自己的根节点
				List<String> keys = curatorFramework.getChildren().forPath(PATH_FOREVER);
				for (String key : keys) {
					String[] nodeKey = key.split("-");
					realNode.put(nodeKey[0], key);
					nodeMap.put(nodeKey[0], Integer.parseInt(nodeKey[1]));
				}
				Integer workerid = nodeMap.get(listenAddress);
				if (workerid != null) {
					//有自己的节点,zk_AddressNode=ip:port
					zkAddressNode = PATH_FOREVER + "/" + realNode.get(listenAddress);
					workerID = workerid;//启动worder时使用会使用
					if (!checkInitTimeStamp(curatorFramework, zkAddressNode)) {
						throw new RuntimeException("init timestamp check error,forever node timestamp gt this node time");
					}
					//准备创建临时节点
					scheduledUploadData(curatorFramework, zkAddressNode);
					updateLocalWorkerID(workerID);
				} else {
					//表示新启动的节点,创建持久节点 ,不用check时间
					String newNode =  curatorFramework
							.create()
							.creatingParentsIfNeeded()
							.withMode(CreateMode.PERSISTENT_SEQUENTIAL)
							.forPath(PATH_FOREVER + "/" + listenAddress + "-", buildData().getBytes());
					zkAddressNode = newNode;
					String[] nodeKey = newNode.split("-");
					workerID = Integer.parseInt(nodeKey[1]);
					scheduledUploadData(curatorFramework, zkAddressNode);
					updateLocalWorkerID(workerID);
				}
			}

		} catch (Exception e) {
			LOGGER.error("init exception", e);
		}

		return true;
	}

	private boolean checkInitTimeStamp(CuratorFramework curator, String zkAddressNode) throws Exception {
		byte[] bytes = curator.getData().forPath(zkAddressNode);
		EndPoint endPoint = JSON.parseObject(new String(bytes), EndPoint.class);
		return !(endPoint.getTimestamp() > System.currentTimeMillis());
	}

	private String buildData() {
		EndPoint endPoint = new EndPoint(ip, port, System.currentTimeMillis());
		return JSON.toJSONString(endPoint);
	}

	private void updateLocalWorkerID(int workerID) {
		File confFile = new File(String.format(PROP_PATH, port));
		boolean exists = confFile.exists();

		if (exists) {
			try {
				FileUtils.writeStringToFile(confFile, "workerID=" + workerID, false);
			} catch (IOException e) {
				LOGGER.error("update file cache error ", e);
			}
		} else {
			try {
				boolean mkdirs = confFile.getParentFile().mkdirs();
				if (mkdirs) {
					if (confFile.createNewFile()) {
						FileUtils.writeStringToFile(confFile, "workerID=" + workerID, false);
						LOGGER.info("local file cache workerID is {}", workerID);
					}
				} else {
				}
			} catch (IOException e) {
				LOGGER.error("craete workerID conf file error", e);
			}
		}
	}

	private void scheduledUploadData(final CuratorFramework curator, final String zkAddressNode) {
		Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				Thread thread = new Thread(r, "schedule-upload-time");
				thread.setDaemon(true);
				return thread;
			}
		}).scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				updateNewData(curator, zkAddressNode);
			}
		}, 1L, 3L, TimeUnit.SECONDS);//每3s上报数据
	}

	private void updateNewData(CuratorFramework curator, String path) {
		try {
			if (System.currentTimeMillis() < lastUpdateTime) {
				return;
			}
			lastUpdateTime = System.currentTimeMillis();
			curator.setData().forPath(path, buildData().getBytes());
		} catch (Exception e) {
			LOGGER.info("update init data error path is {} error is {}", path, e);
		}
	}

	public int getWorkerID() {
		return workerID;
	}

	@Data
	@AllArgsConstructor
	@NoArgsConstructor
	static class EndPoint {
		private String ip;
		private String port;
		private long timestamp;
	}

}
